package com.intct.flink.state;

import com.intct.hbase.bean.WindowBean;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-06-30 15:58
 */
public class BroadcastStateDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setStateBackend(new EmbeddedRocksDBStateBackend());
//        env.setStateBackend(new HashMapStateBackend());

        // 提交作业命令行：flink run-application -y yarn-application -Dstate.backend hashmap -c 程序入口类 jar包

        // flink-conf.yaml文件中state.backend = hashmap

        // 设置并行度
        env.setParallelism(1);

        // 数据源1 业务数据
        /*
        s1,1,1
        s1,1,2
         */
        DataStreamSource<String> sourceDS1 = env.socketTextStream("jd-node", 8888);

        // 转换操作
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS1.map(source -> {
            String[] sources = source.split(",");
            return WindowBean.builder().name(sources[0]).ts(Long.valueOf(sources[1])).vc(Integer.valueOf(sources[2])).build();
        });

        // 数据源2  BroadcastState
        /*
        s1,1
        s2,2
         */
        DataStreamSource<String> sourceDS2 = env.socketTextStream("jd-node", 9999);

        // 定义广播状态描述
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor("msd-1", Types.STRING, Types.INT);
        BroadcastStream<String> broadcastStream = sourceDS2.broadcast(mapStateDescriptor);

        // 数据链接
        BroadcastConnectedStream<WindowBean, String> connectStream = mapDS.connect(broadcastStream);

        SingleOutputStreamOperator<String> processDS = connectStream.process(new BroadcastProcessFunction<WindowBean, String, String>() {

            /**
             * 业务数据处理
             */
            @Override
            public void processElement(WindowBean value, BroadcastProcessFunction<WindowBean, String, String>.ReadOnlyContext ctx,
                                       Collector<String> out) throws Exception {
                // 从上下文件获取广播状态，并且是只读
                ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                // 获取广播状态中特定值  -- 车辆速度超60，为超速
                Integer bVc = broadcastState.get(value.getName()) == null ? 60 : broadcastState.get(value.getName());

                // 如果超预设置60速度
                if (value.getVc() > bVc) {
                    out.collect(value + "车辆速度超过" + bVc + "!");
                }
            }

            /**
             * 广播流处理
             */
            @Override
            public void processBroadcastElement(String value, BroadcastProcessFunction<WindowBean, String, String>.Context ctx,
                                                Collector<String> out) throws Exception {
                String[] valueSplit = value.split(",");
                // 从上下文件中获取广播状态中数据
                BroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                broadcastState.put(valueSplit[0], Integer.valueOf(valueSplit[1]));
            }
        });

        processDS.print();

        env.execute();


    }
}
